package com.intct.func;

import com.alibaba.fastjson.JSONObject;
import com.intct.util.HBaseUtil;
import com.intct.util.RowkeyUtil;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.hadoop.hbase.client.Put;

import java.util.List;
import java.util.Map;

/**
 * @author gufg
 * @since 2025-07-29 10:44
 */
public class LocusSinkHBase implements SinkFunction<String> {

    private String namespace;
    private String tableName;

    private List<String> rowkeys;

    public LocusSinkHBase(String namespace,String tableName, List<String> rowkeys) {
        this.namespace = namespace;
        this.tableName = tableName;
        this.rowkeys = rowkeys;
    }

    @Override
    public void invoke(String jsonStr, Context context) throws Exception {
        /*
        样例数据：
         * {
         *   "driverId": "11",
         *   "lat": "30.70324",
         *   "lng": "104.05355",
         *   "orderId": "17592719043682",
         *   "time": 1475976808
         * }
         *
         * {
         *   "driverId": "end",
         *   "orderId": "17592719043682",
         *   "time": 0
         * }
         */

        System.out.println("开始处理数据......");
        // 转换JSONObject
        JSONObject jsonObj = JSONObject.parseObject(jsonStr);
        String rowkeyStr = new String();
        for (String rowkey : rowkeys) {
            rowkeyStr = rowkeyStr + ":" +jsonObj.getString(rowkey);
        }

        // 根据表分区数量+rowkey值，Rowkey生成
        String rowkeyCode = RowkeyUtil.getRowkey(rowkeyStr, 9);

        System.out.println("rowkeyCode = " + rowkeyCode);

        Put put = new Put(rowkeyCode.getBytes());
        // 字段名称与值
        for (Map.Entry<String, Object> entrys : jsonObj.entrySet()) {
            put.addColumn("info".getBytes(), entrys.getKey().getBytes(), entrys.getValue().toString().getBytes());
        }

        // 创建HBase表
        HBaseUtil.createTable(namespace, tableName, "info");

        // 调用HBase工具类 -- 增修改数据
        HBaseUtil.put(namespace, tableName, put);

        System.out.println("保存数据完成......");
    }
}
